package realtime.ads;

import org.apache.flink.table.api.{Table, TableResult}
import com.example.DwsLayer._
import com.example.DwdLayer._

object AdsLayer {
        // 输出达摩盘基础特征标签表（最终应用层，包含所有基础特征）
        def createAdsDmpBaseFeature(tEnv: StreamTableEnvironment): TableResult = {
        // 注册DWS/DWD汇总表
        createDwsUserAgeSummary(tEnv)
        createDwsUserGenderSummary(tEnv)
        createDwsUserHeightSummary(tEnv)
        createDwsUserWeightSummary(tEnv)
        createDwdUserConstellationDetail(tEnv)
        // 关联所有特征，输出最终标签
        val result = tEnv.executeSql(
        """
        |CREATE TABLE ads_dmp_base_feature (
        |  user_id STRING COMMENT '用户ID（主键）',
        |  age_range STRING COMMENT '年龄标签（18-24/25-29等，文档1-26）',
        |  gender_tag STRING COMMENT '性别标签（female/male/family，文档1-157）',
        |  birth_era STRING COMMENT '出生年代（基于年龄推算，如2001-2007等）',
        |  height_cm INT COMMENT '身高标签（cm，NULL为无效）',
        |  weight_kg DECIMAL(5,1) COMMENT '体重标签（kg，NULL为无效）',
        |  constellation STRING COMMENT '星座标签（未知/白羊座等，文档1-260）',
        |  tag_update_time TIMESTAMP COMMENT '标签更新时间（每日更新，文档1-107）',
        |  PRIMARY KEY (user_id) NOT ENFORCED
        |) WITH (
        |  'connector' = 'hive',
        |  'database-name' = 'ads',
        |  'table-name' = 'ads_dmp_base_feature',
        |  'hive-version' = '3.1.2',
        |  'sink.partition-commit.trigger' = 'process-time',
        |  'sink.partition-commit.delay' = '1 h',
        |  'sink.partition-commit.policy.kind' = 'metastore,success-file'
        |) AS
        |SELECT
        |  COALESCE(a.user_id, b.user_id, c.user_id, d.user_id, e.user_id) AS user_id,
        |  a.final_age_range AS age_range,
        |  b.final_gender_tag AS gender_tag,
        |  -- 基于年龄推算出生年代（当前年份2025，文档未明确，按创建时间默认）
        |  CASE a.final_age_range
        |    WHEN '18-24' THEN '2001-2007'
        |    WHEN '25-29' THEN '1996-2000'
        |    WHEN '30-34' THEN '1991-1995'
        |    WHEN '35-39' THEN '1986-1990'
        |    WHEN '40-49' THEN '1976-1985'
        |    WHEN '50+' THEN '1975及以前'
        |    ELSE '未知'
        |  END AS birth_era,
        |  c.weighted_height AS height_cm,
        |  d.weighted_weight AS weight_kg,
        |  COALESCE(e.constellation, '未知') AS constellation,
        |  CURRENT_TIMESTAMP() AS tag_update_time
        |FROM dws_user_age_summary a
        |FULL JOIN dws_user_gender_summary b ON a.user_id = b.user_id
        |FULL JOIN dws_user_height_summary c ON a.user_id = c.user_id
        |FULL JOIN dws_user_weight_summary d ON a.user_id = d.user_id
        |FULL JOIN dwd_user_constellation_detail e ON a.user_id = e.user_id
        |""".stripMargin)
        result
        }

        // 输出标签测试表（用于验收测试，文档1-264）
        def createAdsDmpFeatureTest(tEnv: StreamTableEnvironment): TableResult = {
        // 注册应用层表
        createAdsDmpBaseFeature(tEnv)
        // 输出10万条以内测试数据（文档1-265）
        val result = tEnv.executeSql(
        """
        |CREATE TABLE ads_dmp_feature_test (
        |  user_id STRING COMMENT '用户ID',
        |  age_range STRING COMMENT '年龄标签',
        |  gender_tag STRING COMMENT '性别标签',
        |  birth_era STRING COMMENT '出生年代',
        |  height_cm INT COMMENT '身高',
        |  weight_kg DECIMAL(5,1) COMMENT '体重',
        |  constellation STRING COMMENT '星座',
        |  test_batch STRING COMMENT '测试批次（当前日期）'
        |) WITH (
        |  'connector' = 'filesystem',
        |  'path' = '/user/hive/warehouse/ads.db/ads_dmp_feature_test',
        |  'format' = 'csv',
        |  'csv.field-delimiter' = ',',
        |  'csv.ignore-parse-errors' = 'true'
        |) AS
        |SELECT
        |  user_id,
        |  age_range,
        |  gender_tag,
        |  birth_era,
        |  height_cm,
        |  weight_kg,
        |  constellation,
        |  DATE_FORMAT(CURRENT_DATE(), 'yyyyMMdd') AS test_batch
        |FROM ads_dmp_base_feature
        |LIMIT 100000  -- 限制10万条以内（文档1-265）
        |""".stripMargin)
        result
        }
        }